// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jaewon

#pragma once

#include <cstdint>
#include <memory>
#include <string>
#include <unordered_map>
#include <vector>

#include "coin1/fastfeed/proto_feed_builder.h"

namespace fastfeed {
namespace detail {

class SwitchingContext {
 public:
  explicit SwitchingContext(int num_workers);
  bool UpdateAndCheck(int worker_idx, int64_t timestamp);
  void SwitchWorker(int worker_idx, ProtoFeedBuilder* builder);

  int active_worker_idx() const { return active_worker_idx_; }

 private:
  const int64_t threshold_ = 10 * 1000000000LL;  // 10 seconds
  int active_worker_idx_ = -1;
  std::vector<int64_t> last_timestamps_;
  ProtoFeedBuilder* active_builder_ = nullptr;
};  // class SwitchingContext

}  // namespace detail

class DoubleShiftFeedBuilder : public FeedBuilderBase {
 public:
  explicit DoubleShiftFeedBuilder(int num_workers);
  DoubleShiftFeedBuilder(int num_workers, const std::string& sub_req);
  ~DoubleShiftFeedBuilder() final = default;

  std::function<void(const proto::Feed&)> GetUpdateFunction(int worker_idx);

  void set_from_timestamp(std::optional<int64_t> from_timestamp) final;

  void Subscribe(
      proto::FeedType feed_type,
      const std::string& custom_feed_type,
      const std::string& symbol,
      const std::string& view_type,
      std::function<void(const view::ViewBase*)> callback) final;

 private:
  void InitBuilders(int num_workers);

  void SetSymbolGroup(const std::string& symbol, const std::string& symbol_group);
  const std::string& FindSymbolGroup(const std::string& symbol);

  detail::SwitchingContext* FindSwitchingContextBySymbol(const std::string& symbol);
  void UpdateByProto(int worker_id, ProtoFeedBuilder* builder, const proto::Feed& feed);

  std::unordered_map<std::string, detail::SwitchingContext*> symbol_to_switching_ctx_;
  std::unordered_map<std::string, std::string> symbol_to_group_;
  std::unordered_map<std::string, std::unique_ptr<detail::SwitchingContext>> switching_ctxs_;

  std::vector<std::unique_ptr<ProtoFeedBuilder>> builders_;
};  // class ProtoFeedBuilder

}  // namespace fastfeed
